package iot.emulator;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import iot.jtt808.DataPack;
import iot.jtt808.FrameDecoder;
import iot.jtt808.terminal.LocationReport;
import iot.jtt808.terminal.TerminalRegister;
import netty.builder.NettyClientBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by dave on 18-7-21 上午11:26.
 */
public class Jtt808Emulator {
    private static final Logger L = LoggerFactory.getLogger(Jtt808Emulator.class);
    private String host;
    private int port;
    private int connections;  // 模拟的连接个数
    private Map<String, Channel> channelMap = new ConcurrentHashMap<>();
    private Random rand = new Random();
    private AtomicInteger seq = new AtomicInteger(0);
    private AtomicLong packCount = new AtomicLong(0);
    private AtomicLong totalBytes = new AtomicLong(0);
    private CountDownLatch connected;

    public Jtt808Emulator(String host, int port, int connections) {
        this.host = host;
        this.port = port;
        this.connections = connections;
        this.connected = new CountDownLatch(connections / 200 * 200);
    }

    public void start() {
        int threadCount = connections / 200;
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < threadCount; i++) {
            final int tid = i;
            Thread th = new Thread(() -> {
                List<String> sid = new ArrayList<>();
                for (int c = 0; c < 200; c++) {
                    String sim = makeSim(c + tid * 200 + c + 1);
                    sid.add(sim);
                    NettyClientBuilder builder = new NettyClientBuilder();
                    builder.channelInitializer(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new FrameDecoder());
                            ch.pipeline().addLast(new EmulatorChannelHandler(sim));
                        }
                    });

                    try {
                        Thread.sleep(20);
                    } catch (InterruptedException e) { }
                    ChannelFuture future = builder.buildAndConnectAsync(host, port);
                    future.addListener(f -> {
                        if (f.isSuccess()) {
                            connected.countDown();
                            L.info(sim + "已连接：" + connected.getCount());
                        } else {
                            L.error(sim + "连接失败");
                        }
                    });
                    channelMap.put(sim, future.channel());
                }

                try {
                    connected.await();
                } catch (InterruptedException e) { }
                L.info("5秒后开始发送数据：" + tid);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) { }

                long ts = System.currentTimeMillis();
                while (!Thread.interrupted()) {
                    for (String s : sid) {
                        Channel ch = channelMap.get(s);
                        byte[] bytes = encodeLocationReport(s);
                        totalBytes.addAndGet(bytes.length);
                        ch.writeAndFlush(Unpooled.wrappedBuffer(bytes)).addListener(f -> {
                            if (!f.isSuccess()) {
                                L.warn("发出位置失败");
                            } else {
                                accumulate(ts);
                            }
                        });
                        bytes = encodeRegistration(s);
                        totalBytes.addAndGet(bytes.length);
                        ch.writeAndFlush(Unpooled.wrappedBuffer(bytes)).addListener(f -> {
                            if (!f.isSuccess()) {
                                L.warn("发出注册失败");
                            } else {
                                accumulate(ts);
                            }
                        });
                    }
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) { }
                }
            });
            threads.add(th);
        }

        L.info("2秒后开始连接");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) { }

        for (Thread th : threads) {
            th.start();
        }

        for (Thread th : threads) {
            try {
                th.join();
            } catch (InterruptedException e) {

            }
        }
    }

    public Map<String, Channel> getChannelMap() {
        return Collections.unmodifiableMap(channelMap);
    }

    private String makeSim(int i) {
        return String.format("0188%08d", i);
    }

    private void accumulate(long start) {
        long c = packCount.addAndGet(1);
        if (c % 20000 == 0) {
            long kb = totalBytes.get() / 1024;
            long tt = System.currentTimeMillis() - start;
            L.info("已发出：" + (c/10000) + "万包, " + kb + "KB, " + tt + "ms. " + (kb * 1024 / tt) + "KB/s");
        }
    }

    private byte[] encodeLocationReport(String sim) {
        DataPack<LocationReport> dp = new DataPack<>();
        dp.setMsgId(0x0200);
        dp.setSim(sim);
        dp.setSequenceNo(nextSeq());

        LocationReport r = new LocationReport();
        r.setAlarmFlag(0);
        r.setFlag(2);
        r.setLat(30 + rand.nextDouble());
        r.setLng(118 + rand.nextDouble());
        r.setHeight(rand.nextInt(1500));
        r.setSpeed(rand.nextInt(800));
        r.setDirection(rand.nextInt(360));
        r.setTime(new Date());
        dp.setBody(r);
        byte[] arr = dp.encode();
         // L.warn("" + arr.length + ": " + javax.xml.bind.DatatypeConverter.printHexBinary(arr));
        return arr;
    }

    private byte[] encodeRegistration(String sim) {
        DataPack<TerminalRegister> dp = new DataPack<>();
        dp.setMsgId(0x0100);
        dp.setSim(sim);
        dp.setSequenceNo(nextSeq());

        TerminalRegister r = new TerminalRegister();
        r.setProvinceId(42);
        r.setCityId(100);
        r.setMakerId("abcde");
        r.setModel("12345678");
        r.setDeviceId("abcdefg");
        r.setColorId(2);
        r.setLicense("鄂A12345");
        dp.setBody(r);
        byte[] arr = dp.encode();
         // L.warn("" + arr.length + ": " + javax.xml.bind.DatatypeConverter.printHexBinary(arr));
        return arr;
    }

    public int nextSeq() {
        int s = seq.addAndGet(1);
        if (s >= 0xFFF) {
            seq.set(1);
            return 1;
        } else {
            return s;
        }
    }
}
